/**
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
 * agreements.  See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership.  The ASF licenses this file to you under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with the License.  You may obtain
 * a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */

package org.apache.zookeeper.server.quorum;

import org.apache.jute.BinaryInputArchive;
import org.apache.jute.BinaryOutputArchive;
import org.apache.jute.Record;
import org.apache.zookeeper.KeeperException.SessionExpiredException;
import org.apache.zookeeper.ZooDefs.OpCode;
import org.apache.zookeeper.server.ByteBufferInputStream;
import org.apache.zookeeper.server.Request;
import org.apache.zookeeper.server.ZooKeeperThread;
import org.apache.zookeeper.server.ZooTrace;
import org.apache.zookeeper.server.quorum.Leader.Proposal;
import org.apache.zookeeper.server.quorum.QuorumPeer.LearnerType;
import org.apache.zookeeper.server.util.SerializeUtils;
import org.apache.zookeeper.server.util.ZxidUtils;
import org.apache.zookeeper.txn.TxnHeader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.util.LinkedList;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock;

import javax.security.sasl.SaslException;

/**
 * Leader和Learner之间都会创建一个这个类的实例。
 * Leader和Follower之间所有的通信都是使用这个类完成的。
 * There will be an instance of this class created by the Leader for each
 * learner. All communication with a learner is handled by this
 * class.
 */
public class LearnerHandler extends ZooKeeperThread {
    private static final Logger LOG = LoggerFactory.getLogger(LearnerHandler.class);

    protected final Socket sock;
    final Leader leader;
    /**
     * 将要发送给Learner服务器的消息。
     */
    final LinkedBlockingQueue<QuorumPacket> queuedPackets = new LinkedBlockingQueue<QuorumPacket>();
    /**
     * If this packet is queued, the sender thread will exit
     */
    final QuorumPacket proposalOfDeath = new QuorumPacket();
    private final BufferedInputStream bufferedInput;
    /**
     * ZooKeeper server identifier of this learner (Follower or Observer)
     */
    protected long sid = 0;
    /**
     * @see LearnerInfo#protocolVersion
     */
    protected int version = 0x1;
    /**Deadline for receiving the next ack. If we are bootstrapping then
     * it's based on the initLimit, if we are done bootstrapping it's based
     * on the syncLimit. Once the deadline is past this learner should
     * be considered no longer "sync'd" with the leader. */
    volatile long tickOfNextAckDeadline;
    private SyncLimitCheck syncLimitCheck = new SyncLimitCheck();

    private BinaryInputArchive ia;
    private BinaryOutputArchive oa;
    private BufferedOutputStream bufferedOutput;

    private LearnerType learnerType = LearnerType.PARTICIPANT;

    LearnerHandler(Socket sock, BufferedInputStream bufferedInput,
                   Leader leader) throws IOException {
        super("LearnerHandler-" + sock.getRemoteSocketAddress());
        this.sock = sock;
        this.leader = leader;
        this.bufferedInput = bufferedInput;
        try {
            leader.self.authServer.authenticate(sock,
                                                new DataInputStream(bufferedInput));
        } catch (IOException e) {
            LOG.error("Server failed to authenticate quorum learner, addr: {}, closing connection",
                      sock.getRemoteSocketAddress(), e);
            try {
                sock.close();
            } catch (IOException ie) {
                LOG.error("Exception while closing socket", ie);
            }
            throw new SaslException("Authentication failure: " + e.getMessage());
        }
    }

    static public String packetToString(QuorumPacket p) {
        String type = null;
        String mess = null;
        Record txn = null;

        switch (p.getType()) {
            case Leader.ACK:
                type = "ACK";
                break;
            case Leader.COMMIT:
                type = "COMMIT";
                break;
            case Leader.FOLLOWERINFO:
                type = "FOLLOWERINFO";
                break;
            case Leader.NEWLEADER:
                type = "NEWLEADER";
                break;
            case Leader.PING:
                type = "PING";
                break;
            case Leader.PROPOSAL:
                type = "PROPOSAL";
                TxnHeader hdr = new TxnHeader();
                try {
                    SerializeUtils.deserializeTxn(p.getData(), hdr);
                    // mess = "transaction: " + txn.toString();
                } catch (IOException e) {
                    LOG.warn("Unexpected exception", e);
                }
                break;
            case Leader.REQUEST:
                type = "REQUEST";
                break;
            case Leader.REVALIDATE:
                type = "REVALIDATE";
                ByteArrayInputStream bis = new ByteArrayInputStream(p.getData());
                DataInputStream dis = new DataInputStream(bis);
                try {
                    long id = dis.readLong();
                    mess = " sessionid = " + id;
                } catch (IOException e) {
                    LOG.warn("Unexpected exception", e);
                }

                break;
            case Leader.UPTODATE:
                type = "UPTODATE";
                break;
            default:
                type = "UNKNOWN" + p.getType();
        }
        String entry = null;
        if (type != null) {
            entry = type + " " + Long.toHexString(p.getZxid()) + " " + mess;
        }
        return entry;
    }

    public Socket getSocket() {
        return sock;
    }

    long getSid() {
        return sid;
    }

    int getVersion() {
        return version;
    }

    @Override
    public String toString() {
        StringBuilder sb = new StringBuilder();
        sb.append("LearnerHandler ").append(sock);
        sb.append(" tickOfNextAckDeadline:").append(tickOfNextAckDeadline());
        sb.append(" synced?:").append(synced());
        sb.append(" queuedPacketLength:").append(queuedPackets.size());
        return sb.toString();
    }

    public LearnerType getLearnerType() {
        return learnerType;
    }

    /**
     * queuedPackets 发送
     *
     * @throws InterruptedException
     */
    private void sendPackets() throws InterruptedException {
        long traceMask = ZooTrace.SERVER_PACKET_TRACE_MASK;
        while (true) {
            try {
                QuorumPacket p;
                // 从队列中非阻塞模式获取一个需要发送的packet；如果为null，则把outputStream中的数据flush出去；然后用阻塞方式获取需要发送的数据packet.
                p = queuedPackets.poll();
                if (p == null) {
                    bufferedOutput.flush();
                    p = queuedPackets.take();
                }
                if (p == proposalOfDeath) {
                    // Packet of death!
                    break;
                }

                if (p.getType() == Leader.PING) {
                    traceMask = ZooTrace.SERVER_PING_TRACE_MASK;
                }
                // 事务请求
                if (p.getType() == Leader.PROPOSAL) {
                    syncLimitCheck.updateProposal(p.getZxid(), System.nanoTime());
                }
                if (LOG.isTraceEnabled()) {
                    ZooTrace.logQuorumPacket(LOG, traceMask, 'o', p);
                }

                // outputStream 发送数据。
                oa.writeRecord(p, "packet");
            } catch (IOException e) {
                if (!sock.isClosed()) {
                    LOG.warn("Unexpected exception at " + this, e);
                    try {
                        // this will cause everything to shutdown on
                        // this learner handler and will help notify
                        // the learner/observer instantaneously
                        sock.close();
                    } catch (IOException ie) {
                        LOG.warn("Error closing socket for handler " + this, ie);
                    }
                }
                break;
            }
        }
    }

    /**
     * This thread will receive packets from the peer and process them and
     * also listen to new connections from new peers.
     */
    @Override
    public void run() {
        try {
            leader.addLearnerHandler(this);
            tickOfNextAckDeadline = leader.self.tick.get() + leader.self.initLimit + leader.self.syncLimit;

            ia = BinaryInputArchive.getArchive(bufferedInput);
            bufferedOutput = new BufferedOutputStream(sock.getOutputStream());
            oa = BinaryOutputArchive.getArchive(bufferedOutput);


            // 收到 OBSERVERINFO 或 FOLLOWERINFO 消息
            QuorumPacket qp = new QuorumPacket();
            ia.readRecord(qp, "packet");

            // 判断数据类型：（Leader收到的第一个packet应该是 FOLLOWERINFO or OBSERVERINFO，否则异常退出此线程）
            if (qp.getType() != Leader.FOLLOWERINFO && qp.getType() != Leader.OBSERVERINFO) {
                LOG.error("First packet " + qp.toString() + " is not FOLLOWERINFO or OBSERVERINFO!");
                return;
            }

            // 根据OBSERVERINFO 或 FOLLOWERINFO 消息数据包 data 字段，解析sid 和 protocolVersion
            byte learnerInfoData[] = qp.getData();
            if (learnerInfoData != null) {
                if (learnerInfoData.length == 8) {
                    this.sid = ByteBuffer.wrap(learnerInfoData).getLong();
                } else {
                    LearnerInfo li = new LearnerInfo();
                    ByteBufferInputStream.byteBuffer2Record(ByteBuffer.wrap(learnerInfoData), li);
                    this.sid = li.getServerid();
                    this.version = li.getProtocolVersion();
                }
            } else {
                this.sid = leader.followerCounter.getAndDecrement();
            }
            LOG.info("Follower sid: " + sid + " : info : " + leader.self.quorumPeers.get(sid));

            // 根据数据被类型，判断当前Learner是 Follower 或是 Observer
            if (qp.getType() == Leader.OBSERVERINFO) {
                learnerType = LearnerType.OBSERVER;
            }

            long lastAcceptedEpoch = ZxidUtils.getEpochFromZxid(qp.getZxid()); // 这是Learner发来最后处理的zxid对应的epoch
            long peerLastZxid;
            StateSummary ss = null;
            long zxid = qp.getZxid();


            // ===================== 协商 newEpoch =====================
            long newEpoch = leader.getEpochToPropose(this.getSid(), lastAcceptedEpoch);

            if (this.getVersion() < 0x10000) {
                // we are going to have to extrapolate the epoch information
                long epoch = ZxidUtils.getEpochFromZxid(zxid);
                ss = new StateSummary(epoch, zxid);
                // fake the message
                leader.waitForEpochAck(this.getSid(), ss);
            } else {

                // =================== LEADERINFO 发送===================
                byte ver[] = new byte[4];
                ByteBuffer.wrap(ver).putInt(0x10000);
                QuorumPacket newEpochPacket = new QuorumPacket(Leader.LEADERINFO, ZxidUtils.makeZxid(newEpoch, 0), ver, null);
                oa.writeRecord(newEpochPacket, "packet");
                bufferedOutput.flush();

                // =================== 读数据 ACKEPOCH ===================
                // ACKEPOCH（Follower -> Leader）包含数据：Follower的currentEpoch（Follower发送时把epoch转成了int类型的）和 lastLoggedZxid
                QuorumPacket ackEpochPacket = new QuorumPacket();
                ia.readRecord(ackEpochPacket, "packet");
                if (ackEpochPacket.getType() != Leader.ACKEPOCH) {
                    LOG.error(ackEpochPacket.toString() + " is not ACKEPOCH");
                    return;
                }
                // 两个参数：peerEpoch 和 zxid
                ss = new StateSummary(ByteBuffer.wrap(ackEpochPacket.getData()).getInt(), ackEpochPacket.getZxid());

                // ==================== 等待 ACKEPOCH ============================
                leader.waitForEpochAck(this.getSid(), ss);
            }
            peerLastZxid = ss.getLastZxid();


            // =================== 数据同步 ===================

            int packetToSend = Leader.SNAP; // SNAP 消息，默认只发给Follower
            long zxidToSend = 0;
            long leaderLastZxid = 0;
            long updates = peerLastZxid; // Follower服务器从哪个zxid开始同步

            // 判断Leader服务器是否需要发送在内存中的提议(proposal)
            ReentrantReadWriteLock lock = leader.zk.getZKDatabase().getLogLock();
            ReadLock rl = lock.readLock();
            try {
                rl.lock();
                final long maxCommittedLog = leader.zk.getZKDatabase().getmaxCommittedLog();
                final long minCommittedLog = leader.zk.getZKDatabase().getminCommittedLog();
                LinkedList<Proposal> proposals = leader.zk.getZKDatabase().getCommittedLog();
                LOG.info("Synchronizing with Follower sid: " + sid + " maxCommittedLog=0x" + Long.toHexString(maxCommittedLog) + " minCommittedLog=0x" + Long.toHexString(minCommittedLog) + " peerLastZxid=0x" + Long.toHexString(peerLastZxid));


                // 如果Follower.lastZxid 和leader.lastZxid相等，说明Follower和Leader的事务相同，则发送empty diff
                if (peerLastZxid == leader.zk.getZKDatabase().getDataTreeLastProcessedZxid()) {
                    LOG.info("leader and follower are in sync, zxid=0x{}", Long.toHexString(peerLastZxid));
                    packetToSend = Leader.DIFF;
                    zxidToSend = peerLastZxid;

                } else if (proposals.size() != 0) {
                    LOG.debug("proposal size is {}", proposals.size());

                    // 如果 peerLastZxid 在 [minCommittedLog, maxCommittedLog] 范围之间，则说明peer存在事务没有处理；则给peer发送‘同步消息’
                    if (minCommittedLog <= peerLastZxid && peerLastZxid <= maxCommittedLog) {
                        LOG.debug("Sending proposals to follower");

                        long prevProposalZxid = minCommittedLog;

                        // 在发送第一个packet之前，我们必须告诉Leader服务器是‘回滚’还是‘同步’
                        boolean firstPacket = true;

                        // 如果上述条件成立，我们可以使用 committedLog 与Follower服务器进行数据同步。我们仅需要确定是否发送‘回滚’消息。
                        packetToSend = Leader.DIFF;
                        zxidToSend = maxCommittedLog;

                        for (Proposal propose : proposals) {
                            // skip the proposals the peer already has
                            if (propose.packet.getZxid() <= peerLastZxid) {
                                prevProposalZxid = propose.packet.getZxid();
                                continue;
                            } else {
                                // 数据同步：找出Follower有但Leader没有的事务，对于这些事务让Follower进行回滚。
                                if (firstPacket) {
                                    firstPacket = false;
                                    // 疑问:什么情况会成立？这个目的是什么？答：详见图片：zookeeper-release-3.4.11\docs\images\数据同步-回滚事务.png
                                    if (prevProposalZxid < peerLastZxid) {
                                        // 在发送数据同步消息之前，先发送一个回滚消息
                                        packetToSend = Leader.TRUNC;
                                        zxidToSend = prevProposalZxid;
                                        updates = zxidToSend;
                                    }
                                }
                                // 疑问:为什么需要发送两个数据包？答：第一个数据包表示事务提议，第二个数据包表示让Follower提交此事务
                                queuePacket(propose.packet);
                                queuePacket(new QuorumPacket(Leader.COMMIT, propose.packet.getZxid(), null, null));
                            }
                        }

                    } else if (peerLastZxid > maxCommittedLog) {
                        // 如果Learner事务id 大于 Learner事务id，则说明Learner的事务需要回滚。
                        LOG.debug("Sending TRUNC to follower zxidToSend=0x{} updates=0x{}", Long.toHexString(maxCommittedLog), Long.toHexString(updates));

                        packetToSend = Leader.TRUNC; // 回滚消息
                        zxidToSend = maxCommittedLog;
                        updates = zxidToSend;
                    } else {
                        LOG.warn("Unhandled proposal scenario");
                    }
                } else {
                    // just let the state transfer happen
                    LOG.debug("proposals is empty");
                }

                LOG.info("Sending " + Leader.getPacketType(packetToSend));
                leaderLastZxid = leader.startForwarding(this, updates);

            } finally {
                rl.unlock();
            }

            // =================== NEWLEADER 消息创建过程（这里只是把消息放到了queuedPackets中，并没有真正发送）===================

            QuorumPacket newLeaderQP = new QuorumPacket(Leader.NEWLEADER,
                                                        ZxidUtils.makeZxid(newEpoch, 0), null, null);
            if (getVersion() < 0x10000) {
                // 当前版本小于 65536，则直接把消息发给Follower服务器
                oa.writeRecord(newLeaderQP, "packet");
            } else {
                // 当前版本大于等于 65536，把 NEWLEADER 消息放到队列中。
                queuedPackets.add(newLeaderQP);
            }
            bufferedOutput.flush();

            // 如果需要‘全量’数据同步
            if (packetToSend == Leader.SNAP) {
                zxidToSend = leader.zk.getZKDatabase().getDataTreeLastProcessedZxid();
            }

            // ******************* 发送数据同步请求 消息类型可能是 snap、diff、trunc *******************
            oa.writeRecord(new QuorumPacket(packetToSend, zxidToSend, null, null), "packet");
            bufferedOutput.flush();

            // 如果是‘全量’数据同步，则把database序列化到outputStream
            if (packetToSend == Leader.SNAP) {
                LOG.info("Sending snapshot last zxid of peer is 0x" + Long.toHexString(peerLastZxid) + "  zxid of leader is 0x" + Long.toHexString(leaderLastZxid) + "sent zxid of db as 0x" + Long.toHexString(zxidToSend));
                // 快照dump to Learner服务器
                leader.zk.getZKDatabase().serializeSnapshot(oa);
                oa.writeString("BenWasHere", "signature");
            }
            bufferedOutput.flush();

            // Start sending packets
            new Thread() {
                @Override
                public void run() {
                    Thread.currentThread().setName("Sender-" + sock.getRemoteSocketAddress());
                    try {
                        // 如果想要知道queuedPackets此时有哪些数据，可在run()方法中是用正则模式搜索：    (?:queuedPackets|queuePacket)
                        sendPackets();
                    } catch (InterruptedException e) {
                        LOG.warn("Unexpected interruption", e);
                    }
                }
            }.start();


            // 上面一段逻辑后，发送的最后一个消息是 NEWLEADER 。

            // ====================== NEWLEADERR ACK ======================

            qp = new QuorumPacket();
            ia.readRecord(qp, "packet");
            if (qp.getType() != Leader.ACK) {
                LOG.error("Next packet was supposed to be an ACK");
                return;
            }
            LOG.info("Received NEWLEADER-ACK message from " + getSid());

            // 等待 （NEWLEADER的）ACK ：表示Learner数据同步完毕。服务器数量达到quorum之后，才继续往下执行。
            leader.waitForNewLeaderAck(getSid(), qp.getZxid(), getLearnerType());

            // syncLimitCheck 不是线程。
            syncLimitCheck.start();

            // 设置超时时间 SO_TIMEOUT
            sock.setSoTimeout(leader.self.tickTime * leader.self.syncLimit);

            // 等待Leader服务器启动完毕
            synchronized (leader.zk) {
                while (!leader.zk.isRunning() && !this.isInterrupted()) {
                    leader.zk.wait(20);
                }
            }

            // Mutation packets will be queued during the serialize,
            // so we need to mark when the peer can actually start using the data
            // ===================== UPTODATE =====================
            queuedPackets.add(new QuorumPacket(Leader.UPTODATE, -1, null, null));


            // ================ Leader服务器启动完毕、数据同步完毕。开始处理事务请求 ================
            while (true) {
                qp = new QuorumPacket();
                ia.readRecord(qp, "packet");

                if (LOG.isTraceEnabled()) {
                    long traceMask = (qp.getType() == Leader.PING) ? ZooTrace.SERVER_PING_TRACE_MASK : ZooTrace.SERVER_PACKET_TRACE_MASK;
                    ZooTrace.logQuorumPacket(LOG, traceMask, 'i', qp);
                }

                // 下一次ack最后期限。   todo:作用：
                tickOfNextAckDeadline = leader.self.tick.get() + leader.self.syncLimit;

                ByteBuffer bb;
                long sessionId;
                int cxid;
                int type;

                switch (qp.getType()) {
                    case Leader.ACK:
                        if (this.learnerType == LearnerType.OBSERVER) {
                            if (LOG.isDebugEnabled()) {
                                LOG.debug("Received ACK from Observer  " + this.sid);
                            }
                        }

                        syncLimitCheck.updateAck(qp.getZxid());
                        leader.processAck(this.sid, qp.getZxid(), sock.getLocalSocketAddress());
                        break;
                    case Leader.PING:
                        // Process the touches
                        ByteArrayInputStream bis = new ByteArrayInputStream(qp.getData());
                        DataInputStream dis = new DataInputStream(bis);
                        while (dis.available() > 0) {
                            long sess = dis.readLong();
                            int to = dis.readInt();
                            leader.zk.touch(sess, to);
                        }
                        break;
                    case Leader.REVALIDATE:
                        bis = new ByteArrayInputStream(qp.getData());
                        dis = new DataInputStream(bis);
                        long id = dis.readLong();
                        int to = dis.readInt();
                        ByteArrayOutputStream bos = new ByteArrayOutputStream();
                        DataOutputStream dos = new DataOutputStream(bos);
                        dos.writeLong(id);
                        boolean valid = leader.zk.touch(id, to);
                        if (valid) {
                            try {
                                //set the session owner as the follower that owns the session
                                leader.zk.setOwner(id, this);
                            } catch (SessionExpiredException e) {
                                LOG.error("Somehow session " + Long.toHexString(id) + " expired right after being renewed! (impossible)", e);
                            }
                        }
                        if (LOG.isTraceEnabled()) {
                            ZooTrace.logTraceMessage(LOG, ZooTrace.SESSION_TRACE_MASK, "Session 0x" + Long.toHexString(id) + " is valid: " + valid);
                        }
                        dos.writeBoolean(valid);
                        qp.setData(bos.toByteArray());
                        queuedPackets.add(qp);
                        break;
                    case Leader.REQUEST:
                        bb = ByteBuffer.wrap(qp.getData());
                        sessionId = bb.getLong();
                        cxid = bb.getInt();
                        type = bb.getInt();
                        bb = bb.slice();
                        Request si;
                        if (type == OpCode.sync) {
                            si = new LearnerSyncRequest(this, sessionId, cxid, type, bb, qp.getAuthinfo());
                        } else {
                            si = new Request(null, sessionId, cxid, type, bb, qp.getAuthinfo());
                        }
                        si.setOwner(this);
                        leader.zk.submitRequest(si);
                        break;
                    default:
                        LOG.warn("unexpected quorum packet, type: {}", packetToString(qp));
                        break;
                }
            } // end  while (true)
        } catch (IOException e) {
            if (sock != null && !sock.isClosed()) {
                LOG.error("Unexpected exception causing shutdown while sock " + "still open", e);
                //close the socket to make sure the
                //other side can see it being close
                try {
                    sock.close();
                } catch (IOException ie) {
                    // do nothing
                }
            }
        } catch (InterruptedException e) {
            LOG.error("Unexpected exception causing shutdown", e);
        } finally {
            LOG.warn("******* GOODBYE " + (sock != null ? sock.getRemoteSocketAddress() : "<null>") + " ********");
            shutdown();
        }
    }

    public void shutdown() {
        // Send the packet of death
        try {
            queuedPackets.put(proposalOfDeath);
        } catch (InterruptedException e) {
            LOG.warn("Ignoring unexpected exception", e);
        }
        try {
            if (sock != null && !sock.isClosed()) {
                sock.close();
            }
        } catch (IOException e) {
            LOG.warn("Ignoring unexpected exception during socket close", e);
        }
        this.interrupt();
        leader.removeLearnerHandler(this);
    }

    public long tickOfNextAckDeadline() {
        return tickOfNextAckDeadline;
    }

    /**
     * Leader服务器给peer服务器发送ping消息。
     *
     * @see Leader#PING PING消息描述
     */
    public void ping() {
        long id;
        // 检查是否超时
        if (syncLimitCheck.check(System.nanoTime())) {
            synchronized (leader) {
                id = leader.lastProposed;
            }
            queuePacket(new QuorumPacket(Leader.PING, id, null, null));
        } else {
            LOG.warn("Closing connection to peer due to transaction timeout.");
            // 如果Leader服务器与Learner服务器超时，则断开与其的连接。
            shutdown();
        }
    }

    void queuePacket(QuorumPacket p) {
        queuedPackets.add(p);
    }

    /**
     * 判断是否与Learner服务器同步
     * 在线程启动 并且
     * @return
     */
    public boolean synced() {
        return isAlive() && leader.self.tick.get() <= tickOfNextAckDeadline;
    }

    /**
     * This class controls the time that the Leader has been
     * waiting for acknowledgement of a proposal from this Learner.
     * If the time is above syncLimit, the connection will be closed.
     * It keeps track of only one proposal at a time, when the ACK for
     * that proposal arrives, it switches to the last proposal received
     * or clears the value if there is no pending proposal.
     */
    private class SyncLimitCheck {
        private boolean started = false;
        private long currentZxid = 0;
        private long currentTime = 0;
        private long nextZxid = 0;
        private long nextTime = 0;

        public synchronized void start() {
            started = true;
        }

        public synchronized void updateProposal(long zxid, long time) {
            if (!started) {
                return;
            }
            if (currentTime == 0) {
                currentTime = time;
                currentZxid = zxid;
            } else {
                nextTime = time;
                nextZxid = zxid;
            }
        }

        public synchronized void updateAck(long zxid) {
            if (currentZxid == zxid) {
                currentTime = nextTime;
                currentZxid = nextZxid;
                nextTime = 0;
                nextZxid = 0;
            } else if (nextZxid == zxid) {
                LOG.warn("ACK for " + zxid + " received before ACK for " + currentZxid + "!!!!");
                nextTime = 0;
                nextZxid = 0;
            }
        }


        /**
         * 检查距离上一次通信，是否超时。
         *
         * @param time 当前系统时间
         * @return
         */
        public synchronized boolean check(long time) {
            if (currentTime == 0) {
                return true;
            } else {
                long msDelay = (time - currentTime) / 1000000;
                return msDelay < leader.self.tickTime * leader.self.syncLimit;
            }
        }
    }
}
